package app.dws;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import app.bean.CategoryAnalysisAccumulator;
import app.bean.CategoryMiningResult;
import app.bean.MarketProduct;
import app.function.CategoryMiningProcessFunction;
import utils.MyClickHouseUtil;
import utils.MyKafkaUtil;
import utils.SystemConfigUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;

/**
 * 品类挖掘处理器 - 符合实时数仓设计规范
 * 实现标准化的DWS层数据处理流程
 */
public class DwsMarketCategoryMining {
    
    private static final Logger logger = LoggerFactory.getLogger(DwsMarketCategoryMining.class);
    private static final String DATA_SOURCE = "kafka_dwd_market_product";
    private static final Integer VERSION = 1;
    
    public static void main(String[] args) throws Exception {
        // 初始化系统配置，避免Hadoop路径问题
        SystemConfigUtil.initializeSystemConfig();
        
        logger.info("启动DwsMarketCategoryMining...");
        // 创建流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        
        // 从Kafka读取数据
        String topic = "dwd_market_product";
        String groupId = "dws_market_category_mining_group";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
        
        // 转换为MarketProduct对象并添加水位线
        SingleOutputStreamOperator<MarketProduct> productDS = kafkaDS.map(new MapFunction<String, MarketProduct>() {
            @Override
            public MarketProduct map(String jsonStr) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(jsonStr);
                MarketProduct product = new MarketProduct();
                product.setProductId(jsonObject.getString("product_id"));
                product.setProductName(jsonObject.getString("product_name"));
                product.setCategoryId(jsonObject.getString("category_id"));
                product.setCategoryName(jsonObject.getString("category_name"));
                product.setBrandId(jsonObject.getString("brand_id"));
                product.setBrandName(jsonObject.getString("brand_name"));
                product.setPrice(jsonObject.getDouble("price"));
                product.setShopId(jsonObject.getString("shop_id"));
                product.setShopName(jsonObject.getString("shop_name"));
                product.setSalesVolume(jsonObject.getInteger("sales_volume"));
                product.setSalesAmount(jsonObject.getDouble("sales_amount"));
                product.setVisitorCount(jsonObject.getInteger("visitor_count"));
                product.setConversionRate(jsonObject.getDouble("conversion_rate"));
                return product;
            }
        }).assignTimestampsAndWatermarks(
            WatermarkStrategy.<MarketProduct>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner<MarketProduct>() {
                    @Override
                    public long extractTimestamp(MarketProduct element, long recordTimestamp) {
                        return System.currentTimeMillis();
                    }
                })
        );
        
        // 按品类ID分组
        KeyedStream<MarketProduct, String> keyedStream = productDS.keyBy(new KeySelector<MarketProduct, String>() {
            @Override
            public String getKey(MarketProduct value) throws Exception {
                return value.getCategoryId();
            }
        });
        
        // 1. 计算品类机会挖掘
        SingleOutputStreamOperator<CategoryMiningResult> opportunityMiningDS = keyedStream
            .window(TumblingEventTimeWindows.of(Time.days(7)))
            .aggregate(new CategoryOpportunityAggregateFunction(), new CategoryOpportunityWindowFunction())
            .keyBy(mining -> "global")
            .process(new CategoryMiningProcessFunction("opportunity", 50));
        
        // 2. 计算品类增长挖掘
        SingleOutputStreamOperator<CategoryMiningResult> growthMiningDS = keyedStream
            .window(TumblingEventTimeWindows.of(Time.days(7)))
            .aggregate(new CategoryGrowthAggregateFunction(), new CategoryGrowthWindowFunction())
            .keyBy(mining -> "global")
            .process(new CategoryMiningProcessFunction("growth", 50));
        
        // 3. 计算品类市场份额挖掘
        SingleOutputStreamOperator<CategoryMiningResult> marketShareMiningDS = keyedStream
            .window(TumblingEventTimeWindows.of(Time.days(7)))
            .aggregate(new CategoryMarketShareAggregateFunction(), new CategoryMarketShareWindowFunction())
            .keyBy(mining -> "global")
            .process(new CategoryMiningProcessFunction("market_share", 50));
        
        // 4. 计算品类供需挖掘
        SingleOutputStreamOperator<CategoryMiningResult> supplyDemandMiningDS = keyedStream
            .window(TumblingEventTimeWindows.of(Time.days(7)))
            .aggregate(new CategorySupplyDemandAggregateFunction(), new CategorySupplyDemandWindowFunction())
            .keyBy(mining -> "global")
            .process(new CategoryMiningProcessFunction("supply_demand", 50));
        
        // 将结果写入ClickHouse - 使用符合实时数仓规范的SQL
        opportunityMiningDS.addSink(MyClickHouseUtil.getSinkFunction(
            "INSERT INTO dws_category_opportunity_mining VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        ));
        
        growthMiningDS.addSink(MyClickHouseUtil.getSinkFunction(
            "INSERT INTO dws_category_growth_mining VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        ));
        
        marketShareMiningDS.addSink(MyClickHouseUtil.getSinkFunction(
            "INSERT INTO dws_category_market_share_mining VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        ));
        
        supplyDemandMiningDS.addSink(MyClickHouseUtil.getSinkFunction(
            "INSERT INTO dws_category_supply_demand_mining VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        ));
        
        // 执行任务
        env.execute("Category Mining Job");
    }
    
    // 品类机会挖掘聚合函数
    public static class CategoryOpportunityAggregateFunction implements AggregateFunction<MarketProduct, CategoryAnalysisAccumulator, CategoryAnalysisAccumulator> {
        @Override
        public CategoryAnalysisAccumulator createAccumulator() {
            return new CategoryAnalysisAccumulator();
        }
        
        @Override
        public CategoryAnalysisAccumulator add(MarketProduct value, CategoryAnalysisAccumulator accumulator) {
            accumulator.categoryId = value.getCategoryId();
            accumulator.categoryName = value.getCategoryName();
            accumulator.totalSalesAmount += value.getSalesAmount();
            accumulator.totalSalesCount += value.getSalesVolume();
            accumulator.productCount++;
            
            // 统计品牌和店铺数量
            accumulator.brandCounts.merge(value.getBrandName(), 1L, Long::sum);
            accumulator.shopCounts.merge(value.getShopName(), 1L, Long::sum);
            
            accumulator.count++;
            return accumulator;
        }
        
        @Override
        public CategoryAnalysisAccumulator getResult(CategoryAnalysisAccumulator accumulator) {
            return accumulator;
        }
        
        @Override
        public CategoryAnalysisAccumulator merge(CategoryAnalysisAccumulator a, CategoryAnalysisAccumulator b) {
            a.totalSalesAmount += b.totalSalesAmount;
            a.totalSalesCount += b.totalSalesCount;
            a.productCount += b.productCount;
            
            b.brandCounts.forEach((k, v) -> a.brandCounts.merge(k, v, Long::sum));
            b.shopCounts.forEach((k, v) -> a.shopCounts.merge(k, v, Long::sum));
            
            a.count += b.count;
            return a;
        }
    }
    
    // 品类机会挖掘窗口函数
    public static class CategoryOpportunityWindowFunction implements WindowFunction<CategoryAnalysisAccumulator, CategoryMiningResult, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<CategoryAnalysisAccumulator> input, Collector<CategoryMiningResult> out) throws Exception {
            CategoryAnalysisAccumulator accumulator = input.iterator().next();
            
            CategoryMiningResult result = new CategoryMiningResult();
            // 设置业务字段
            result.setCategoryId(accumulator.categoryId);
            result.setCategoryName(accumulator.categoryName);
            
            // 设置基础度量字段
            result.setTotalSalesAmount(accumulator.totalSalesAmount);
            result.setTotalSalesCount(accumulator.totalSalesCount);
            result.setProductCount(accumulator.productCount);
            result.setBrandCount((long)accumulator.brandCounts.size());
            result.setShopCount((long)accumulator.shopCounts.size());
            
            // 计算平均价格
            if (accumulator.totalSalesCount > 0) {
                result.setAvgPrice(accumulator.totalSalesAmount / accumulator.totalSalesCount);
            }
            
            // 计算竞争强度
            if (accumulator.productCount > 0) {
                result.setCompetitorCount(accumulator.shopCounts.size());
                result.setCompetitionIntensity((double)accumulator.shopCounts.size() / accumulator.productCount);
            }
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            result.setWindowTime(sdf.format(new Date(window.getStart())), sdf.format(new Date(window.getEnd())));
            result.setTs(window.getEnd());
            
            out.collect(result);
        }
    }
    
    // 品类增长挖掘聚合函数
    public static class CategoryGrowthAggregateFunction implements AggregateFunction<MarketProduct, CategoryAnalysisAccumulator, CategoryAnalysisAccumulator> {
        @Override
        public CategoryAnalysisAccumulator createAccumulator() {
            return new CategoryAnalysisAccumulator();
        }
        
        @Override
        public CategoryAnalysisAccumulator add(MarketProduct value, CategoryAnalysisAccumulator accumulator) {
            accumulator.categoryId = value.getCategoryId();
            accumulator.categoryName = value.getCategoryName();
            accumulator.totalSalesAmount += value.getSalesAmount();
            accumulator.totalSalesCount += value.getSalesVolume();
            accumulator.productCount++;
            accumulator.count++;
            return accumulator;
        }
        
        @Override
        public CategoryAnalysisAccumulator getResult(CategoryAnalysisAccumulator accumulator) {
            return accumulator;
        }
        
        @Override
        public CategoryAnalysisAccumulator merge(CategoryAnalysisAccumulator a, CategoryAnalysisAccumulator b) {
            a.totalSalesAmount += b.totalSalesAmount;
            a.totalSalesCount += b.totalSalesCount;
            a.productCount += b.productCount;
            a.count += b.count;
            return a;
        }
    }
    
    // 品类增长挖掘窗口函数
    public static class CategoryGrowthWindowFunction implements WindowFunction<CategoryAnalysisAccumulator, CategoryMiningResult, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<CategoryAnalysisAccumulator> input, Collector<CategoryMiningResult> out) throws Exception {
            CategoryAnalysisAccumulator accumulator = input.iterator().next();
            
            CategoryMiningResult result = new CategoryMiningResult();
            // 设置业务字段
            result.setCategoryId(accumulator.categoryId);
            result.setCategoryName(accumulator.categoryName);
            
            // 设置基础度量字段
            result.setTotalSalesAmount(accumulator.totalSalesAmount);
            result.setTotalSalesCount(accumulator.totalSalesCount);
            result.setProductCount(accumulator.productCount);
            
            // 模拟设置当前和上期销售数据
            result.setCurrentSales(accumulator.totalSalesAmount);
            result.setPreviousSales(accumulator.totalSalesAmount * 0.8); // 模拟上期数据
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            result.setWindowTime(sdf.format(new Date(window.getStart())), sdf.format(new Date(window.getEnd())));
            result.setTs(window.getEnd());
            
            out.collect(result);
        }
    }
    
    // 品类市场份额挖掘聚合函数
    public static class CategoryMarketShareAggregateFunction implements AggregateFunction<MarketProduct, CategoryAnalysisAccumulator, CategoryAnalysisAccumulator> {
        @Override
        public CategoryAnalysisAccumulator createAccumulator() {
            return new CategoryAnalysisAccumulator();
        }
        
        @Override
        public CategoryAnalysisAccumulator add(MarketProduct value, CategoryAnalysisAccumulator accumulator) {
            accumulator.categoryId = value.getCategoryId();
            accumulator.categoryName = value.getCategoryName();
            accumulator.totalSalesAmount += value.getSalesAmount();
            accumulator.totalSalesCount += value.getSalesVolume();
            accumulator.productCount++;
            accumulator.count++;
            return accumulator;
        }
        
        @Override
        public CategoryAnalysisAccumulator getResult(CategoryAnalysisAccumulator accumulator) {
            return accumulator;
        }
        
        @Override
        public CategoryAnalysisAccumulator merge(CategoryAnalysisAccumulator a, CategoryAnalysisAccumulator b) {
            a.totalSalesAmount += b.totalSalesAmount;
            a.totalSalesCount += b.totalSalesCount;
            a.productCount += b.productCount;
            a.count += b.count;
            return a;
        }
    }
    
    // 品类市场份额挖掘窗口函数
    public static class CategoryMarketShareWindowFunction implements WindowFunction<CategoryAnalysisAccumulator, CategoryMiningResult, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<CategoryAnalysisAccumulator> input, Collector<CategoryMiningResult> out) throws Exception {
            CategoryAnalysisAccumulator accumulator = input.iterator().next();
            
            CategoryMiningResult result = new CategoryMiningResult();
            // 设置业务字段
            result.setCategoryId(accumulator.categoryId);
            result.setCategoryName(accumulator.categoryName);
            
            // 设置基础度量字段
            result.setTotalSalesAmount(accumulator.totalSalesAmount);
            result.setTotalSalesCount(accumulator.totalSalesCount);
            result.setProductCount(accumulator.productCount);
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            result.setWindowTime(sdf.format(new Date(window.getStart())), sdf.format(new Date(window.getEnd())));
            result.setTs(window.getEnd());
            
            out.collect(result);
        }
    }
    
    // 品类供需挖掘聚合函数
    public static class CategorySupplyDemandAggregateFunction implements AggregateFunction<MarketProduct, CategoryAnalysisAccumulator, CategoryAnalysisAccumulator> {
        @Override
        public CategoryAnalysisAccumulator createAccumulator() {
            return new CategoryAnalysisAccumulator();
        }
        
        @Override
        public CategoryAnalysisAccumulator add(MarketProduct value, CategoryAnalysisAccumulator accumulator) {
            accumulator.categoryId = value.getCategoryId();
            accumulator.categoryName = value.getCategoryName();
            accumulator.totalSalesAmount += value.getSalesAmount();
            accumulator.totalSalesCount += value.getSalesVolume();
            accumulator.productCount++;
            accumulator.count++;
            return accumulator;
        }
        
        @Override
        public CategoryAnalysisAccumulator getResult(CategoryAnalysisAccumulator accumulator) {
            return accumulator;
        }
        
        @Override
        public CategoryAnalysisAccumulator merge(CategoryAnalysisAccumulator a, CategoryAnalysisAccumulator b) {
            a.totalSalesAmount += b.totalSalesAmount;
            a.totalSalesCount += b.totalSalesCount;
            a.productCount += b.productCount;
            a.count += b.count;
            return a;
        }
    }
    
    // 品类供需挖掘窗口函数
    public static class CategorySupplyDemandWindowFunction implements WindowFunction<CategoryAnalysisAccumulator, CategoryMiningResult, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<CategoryAnalysisAccumulator> input, Collector<CategoryMiningResult> out) throws Exception {
            CategoryAnalysisAccumulator accumulator = input.iterator().next();
            
            CategoryMiningResult result = new CategoryMiningResult();
            // 设置业务字段
            result.setCategoryId(accumulator.categoryId);
            result.setCategoryName(accumulator.categoryName);
            
            // 设置基础度量字段
            result.setTotalSalesAmount(accumulator.totalSalesAmount);
            result.setTotalSalesCount(accumulator.totalSalesCount);
            result.setProductCount(accumulator.productCount);
            
            // 设置供需数据（模拟）
            result.setDemandCount(accumulator.totalSalesCount * 2); // 模拟需求量
            result.setSupplyCount(accumulator.productCount);
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            result.setWindowTime(sdf.format(new Date(window.getStart())), sdf.format(new Date(window.getEnd())));
            result.setTs(window.getEnd());
            
            out.collect(result);
        }
    }
}